package rdd.req;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Serializable;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class Spark79_HotCategoryTop10 {
    public static void main(String[] args) {
        final SparkConf conf = new SparkConf();
        conf.setMaster("local[*]");
        conf.setAppName("HotCategoryTop10");
        final JavaSparkContext jsc = new JavaSparkContext(conf);
        final JavaRDD<String> dataRDD = jsc.textFile("data/user_visit_action.txt");

        //1.将是搜索行为的数据筛选掉
        final JavaRDD<String> filterRDD = dataRDD.filter(
                line -> {
                    final String[] ss = line.split("\t");
                    return "\\N".equals(ss[5]);
                }
        );
        //将筛选出来的数据进行分组统计，因为在点击的品类只能是一种可以无所谓，但是下单和购买的品类可以是很多种
        //但是此时我们需要将每一行数据变成（品类，1,0,0）或者（品类，0,1,0）或者（品类，0,0,1），但是下单和购买的品类可以是多种
        //需要多品类名进行拆分，所以需要使用flatmap而不是map


        final JavaRDD<HotCategory> flatMapRDD = filterRDD.flatMap(
                line ->{
                    final String[] ss = line.split("\t");
                    //因为涉及到点击、下单、购买的格式不同，需要进行判断是哪个类型的行为
                    if(!"-1".equals(ss[6])){                                    //如果点击显示的不是-1，说明是点击行为，格式为（品类，1,0,0）
                        return  Arrays.asList(new HotCategory(ss[6],1L,0L,0L)).iterator();
                    }else if (!"\\N".equals(ss[8])){                           //如果下单不显示为null，说明是下单行为，格式为（品类，0,1,0）
                        //return  Arrays.asList(new HotCategory(ss[8],0L,0L,1L)).iterator();
                        //直接这样写会到导致并没有将品名分开，因为下单和购买的品类可以使多种的，所以表示品类的id字符串是多个品类的拼接字符擦混
                        final String[] ids = ss[8].split(",");
                        List<HotCategory> objs = new ArrayList<>();
                        for (String id : ids){
                            objs.add(new HotCategory(id,0L,1L,0L));
                        }
                        return objs.iterator() ;
                    }else{
                        //return  Arrays.asList(new HotCategory(ss[10],0L,0L,1L)).iterator();
                        final String[] ids = ss[10].split(",");
                        List<HotCategory> objs = new ArrayList<>();
                        for (String id : ids){
                            objs.add(new HotCategory(id,0L,1L,0L));
                        }
                        return objs.iterator() ;
                    }
                }
        );

        final JavaPairRDD<String,HotCategory> kvRDD = flatMapRDD.mapToPair(obj -> new Tuple2<>(obj.getId(),obj));
        //（品类，（品类，1,0,0））
        //（品类，（品类，0,1,0））
        //（品类，（品类，0,0,1））

        JavaPairRDD<String, HotCategory> objCountRDD = kvRDD.reduceByKey(
                (obj1, obj2) -> {
                    obj1.setClickCount(obj1.getClickCount() + obj2.getClickCount());
                    obj1.setOrderCount(obj1.getOrderCount() + obj2.getOrderCount());
                    obj1.setPayCount(obj1.getPayCount() + obj2.getPayCount());
                    return obj1;
                }
        );

        objCountRDD.take(10).forEach(System.out::println);

        jsc.close();
    }
}

class HotCategory  implements Serializable {
    private String id;
    private Long clickCount;
    private Long orderCount;
    private Long PayCount;

    public HotCategory(String id, Long clickCount, Long orderCount, Long payCount) {
        this.id = id;
        this.clickCount = clickCount;
        this.orderCount = orderCount;
        PayCount = payCount;
    }

    public HotCategory(){

    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public Long getClickCount() {
        return clickCount;
    }

    public void setClickCount(Long clickCount) {
        this.clickCount = clickCount;
    }

    public Long getOrderCount() {
        return orderCount;
    }

    public void setOrderCount(Long orderCount) {
        this.orderCount = orderCount;
    }

    public Long getPayCount() {
        return PayCount;
    }

    public void setPayCount(Long payCount) {
        PayCount = payCount;
    }

    @Override
    public String toString() {
        return "HotCategory{" +
                "id='" + id + '\'' +
                ", clickCount=" + clickCount +
                ", orderCount=" + orderCount +
                ", PayCount=" + PayCount +
                '}';
    }
}
